package org.niit.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}

object TransformRDD3 {
  //转换算子
  def main(args: Array[String]): Unit = {
    //准备环境
    // -- 创建Spark的配置                             CPU核数等于分区数
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark") //重点
    //创建SparkContext 去加载配置
    val sparkContext = new SparkContext(sparkConf)//重点
    //设置日志级别
    sparkContext.setLogLevel("ERROR")

    /*
     1.map：将处理的数据逐条进行映射转换，这里的转换可以是类型转换，也可以是值转换
     需求：将rdd中 1 2 3 4  转换为 2 4 6 8
     */
    val rdd1: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4))
    val resRdd1: RDD[Int] = rdd1.map(_ * 2) //一个rdd转换后 会返回一个新的rdd
    printRDD(resRdd1)

    /*
     2.mapPartitions:将待处理的数据以分区为单位发送到计算节点进行处理。这里的处理是指可以进行任意的处理，哪怕是过滤
     */
    val rdd2: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4),2)//【1，2】 【3，4】
    // 需求：找到每个分区的最大值
    val resRdd2: RDD[Int] = rdd2.mapPartitions(item => {
      //item 指的就是每个分区item1：【1，2】 item2：【3，4】
      List(item.max).iterator
    })
    printRDD(resRdd2)

   /*
    3. flatMap: 先映射 再扁平化 (将两个子列表 合并一个列表)
    */
    val rdd3: RDD[List[Int]] = sparkContext.makeRDD(List(List(1, 2), List(3, 4)))// (1,2),(3,4)==>(1,2,3,4)
    val resRdd3: RDD[Int] = rdd3.flatMap(item => {
      //item 是每个子列表 。由于不需要拆分 ，不需要做其他的操作，可以返回item,直接进行了扁平化的操作
      item
    })
    printRDD(resRdd3)
    //需求：以空格拆分单词
    val rdd4: RDD[String] = sparkContext.makeRDD(List("Hello Spark", "Hello Hadoop"))//==>(Hello,Spark,Hello,Hadoop)
    val resRdd4: RDD[String] = rdd4.flatMap(_.split(" "))
    //_ : "Hello Spark"==["Hello","Spark"]  "Hello Hadoop" = ["Hello","Hadoop"] List( ["Hello","Spark"],  ["Hello","Hadoop"]) ==>(Hello,Spark,Hello,Hadoop)
    val strings: Array[String] = resRdd4.collect()
    println(strings.mkString(","))

    //4.glom:将同一个分区的数据直接转换为相同类型的内存数组进行处理，且分区不变
    val rdd5: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4),2)// 【1，2】   【3，4】
    val resRdd5: RDD[Array[Int]] = rdd5.glom()//将内存中的分区体现为两个数组形式 RDD( Array(1,2),Array(3,4)  )
    val array: Array[Array[Int]] = resRdd5.collect()
    array.foreach(data=> println(data.mkString(",")) )
    //需求：求两个分区最大值的和
    //1.将带有两个区的RDD 转换为 带有两个数组的RDD
    val resRdd6: RDD[Array[Int]] = rdd5.glom()
    //2.对带有两个数组的RDD进行映射得到两个数组中的最大值，并保存到RDD中
    val resRdd7: RDD[Int] = resRdd6.map(array => array.max)//RDD（2，4）
    //3.将保存两个数组的最大值的RDD转换为数组进行求和
    val sum: Int = resRdd7.collect().sum
    println(sum)
    /*
      map(带返回值 循环遍历) 和 foreach(不带返回值 循环遍历)
     */

    /*
      5. groupBy：将数据根据指定的规则进行分组，分区默认不变。但是数据会被打乱重新组合，我们将这样的操作叫做shuffle。
                  极限情况下，数据可能被分在同一个分区中，一个组的数据在一个分区中。但是并不是说一个分区只有一个组
     */
    val rdd6: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4), 2)
    //需求：奇数一组  偶数一组  [1,3]  [2,4]
    val resRdd8: RDD[(Int, Iterable[Int])] = rdd6.groupBy(num => {
      //分组的条件  当余数为0是一组 余数为1的是一组
      num % 2
    })
    resRdd8.collect().foreach( println)
    //需求：按照单词的首字母进行分组
    val rdd7: RDD[String] = sparkContext.makeRDD(List("Hello", "Spark", "HBase", "Scala", "Hadoop", "Python"))
    val resRdd9: RDD[(Char, Iterable[String])] = rdd7.groupBy(_.charAt(0)) // _(0)
    resRdd9.collect().foreach( println )

    /*
      6.fliter:将数据根据指定规则进行筛选过滤，符合规则[true]的数据保留[返回],不符合规则[false]的数据就舍弃。
                当数据筛选后，分区不变，但是分区内的数据可能不均衡，生产环境下，可能会出现数据倾斜
      需求：获得列表中的奇数
     */
    val rdd8: RDD[Int] = sparkContext.makeRDD((1 to 10).toList)
    val resRdd10: RDD[Int] = rdd8.filter(_ % 2 != 0)
    printRDD(resRdd10)

    /*
      7.distinct:将数据集中的重复数据进行去重
     */
    val rdd9: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4, 1, 2, 3, 4, 5))
    val resRdd11: RDD[Int] = rdd9.distinct(2)//去重可以重新分区，如果填了分区数则重分区。如果不填按照默认分区执行
    println("分区数：",resRdd11.getNumPartitions)
    printRDD(resRdd11)

    /*
      8.sortBy：默认是升序，如果想变成降序，那么第二个参数设置为false
     */
    val resRdd12: RDD[Int] = resRdd11.sortBy(num => num,false)
    printRDD(resRdd12)
    //需求： 按照Key进行排序【升序】
    val rdd10: RDD[(String, String)] = sparkContext.makeRDD(List( ("1", "a"), ("3", "c"), ("4", "d"), ("2", "b")))
    var resRdd13: RDD[(String, String)] = rdd10.sortByKey() //方式一
    resRdd13 = rdd10.sortBy( _._1) //方式二
    resRdd13.collect().foreach( println )

    /*
      9.coalesce : 缩减分区。用于大数据集过滤后，提高小数据集 的执行效率
                 当Spark程序中，存在过多小任务的时候，可以通过coalesce方法，收缩合并分区。
                 减少分区的个数，减小任务调度的成本
     */
    val rdd11: RDD[Int] = sparkContext.makeRDD((1 to 6).toList, 3)
    /*
      理想情况下 ：【1，2】 【3，4】，【5，6】==>【1，2，3】 【4，5，6】
       极端情况下：      【1，2，3，5，6】  【 4 】
       为了避免减少分区发生的数据倾斜的问题，可以将coalesce第二个参数设置为true，只能避免数据倾斜的问题
     */
    val resRdd14: RDD[Int] = rdd11.coalesce(2, true)
    //resRdd14.saveAsTextFile("output4")

    /*
      10.repatition : 该算子的内部操作其实就是coalesce操作。参数shuffle默认为true。
                无论将分区数增加还是较少，都要经过shuffle，repatition都可以完成
     */
    val rdd12: RDD[Int] = sparkContext.makeRDD((1 to 6).toList, 3)
    val resRdd15: RDD[Int] = rdd12.repartition(5)
    //resRdd15.saveAsTextFile("output5")

    /*
      11. 交集 并集 差集 拉链
     */
    val rdd13: RDD[Int] = sparkContext.makeRDD((1 to 4).toList)
    val rdd14: RDD[Int] = sparkContext.makeRDD((3 to 6).toList)

    //并集
    val resRdd16: RDD[Int] = rdd13.union(rdd14)
    printRDD(resRdd16)

    //交集
    val resRdd17: RDD[Int] = rdd13.intersection(rdd14)
    printRDD(resRdd17)

    //差集
    val resRdd18: RDD[Int] = rdd13.subtract(rdd14)
    printRDD(resRdd18)

    //拉链
    val resRdd19: RDD[(Int, Int)] = rdd13.zip(rdd14)
    println(resRdd19.collect().mkString(","))

    //12.partitionBy: 将数据按照指定的分区规则(分区器)进行分区。Spark默认的分区器是HashPartitioner
    val rdd15: RDD[Int] = sparkContext.makeRDD((1 to 4).toList, 2)//【1，2】  【3，4】
    //partitionBy 使用的规则 必须是 KV形式的数据         [1,2,3,4]  ==>  [ (1,1),(2,1),(3,1),(4,1)  ]
    //隐式转换  map(遍历循环 有返回值 一般转换数据的操作) 和 foreach(遍历循环  没有返回值)
    val mapRdd: RDD[(Int, Int)] = rdd15.map((_, 1)) //==>[ (1,1),(2,1),(3,1),(4,1)  ]
    val resRdd20: RDD[(Int, Int)] = mapRdd.repartition(3)
    //resRdd20.saveAsTextFile("output6")

    val resRdd21: RDD[(Int, Int)] = mapRdd.partitionBy(new HashPartitioner(3))
    //resRdd21.saveAsTextFile("output7")


    //13.reduceByKey:将相同Key的数据进行聚合
    val rdd16: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 1), ("a", 2), ("b", 3), ("b", 4)))
    /*
         key:a  value:[1,2]
         key:b  value:[3,4]
         a:1+2 =3
         b:3+4 =7
     */
    var resRdd22: RDD[(String, Int)] = rdd16.reduceByKey((x: Int, y: Int) => {
      x + y // a ： 1+2  b:3 + 7
    })
    resRdd22 = rdd16.reduceByKey(_ + _)
    resRdd22.collect().foreach( println )

    //14.groupByKey: 将相同Key对Value进行分组
    val resRdd23: RDD[(String, Iterable[Int])] = rdd16.groupByKey()
    // [(a,[1,2], (b,[3,4]) ]
    resRdd23.collect().foreach( println )

    //14.aggregateByKey:根据不同的规则进行分区内计算 和 分区间计算
    /*
        [ 【(a,1) (a,2) (b,3),(a,0) (b,0)】   【(b,4) (b,5) (a,6) (a,0) (b,0)】  ]  先找到分区内的最大值，然后进行分区间相加
        第一个分区： 最大值 根据Key来找最大值的  (a,2)   (b,3)
        第二个分区： 最大值 根据Key来找最大值的  (b,5)   (a,6)
        a 2 + 6
        b 3 + 5
        (a,8) (b,8)
     */
    val rdd17: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 1), ("a", 2), ("b", 3), ("b", 4), ("b", 5), ("a", 6)), 2)
    val resRdd24: RDD[(String, Int)] = rdd17.aggregateByKey(0)(// rdd17中元素分区内添加默认值(a,0) (b,0)
      //计算规则
      (x, y) => math.max(x, y), //分区内的规则 x和y指的是相同Key的值进行比较
      (x, y) => x + y //分区间的规则  相同Key进行相加
    )
    resRdd24.collect().foreach( println )
    //需求：求两个分区间相同Key下Value的总和
    var resRdd25: RDD[(String, Int)] = rdd17.aggregateByKey(0)(
      //计算规则
      (x, y) => x + y,
      (x, y) => x + y
    )
    resRdd25.collect().foreach( println )
    //简写方式
    resRdd25 = rdd17.aggregateByKey(0)( _+_, _+_)
    //如果分区内和分区间的计算规则是一样 可以直接使用foldByKey
    resRdd25 = rdd17.foldByKey(0)(_ + _)


    //aggregateByKey 计算平均值  ("a", 1), ("a", 2), ("b", 3), ("b", 4), ("b", 5), ("a", 6)   a=（1 + 2 + 6）/3   b= (3+4+5)/3
                                  // x 可以是任意类型和任意值 x = (0,0)
                                // (a,x) (b,0x)                       (a,x) (b,x)
    /*
      (0,0)==t：第一个0表示的是 v 的初始值相加 a : 0+ 1 + 2  b:0 + 3
      (0,0)==t：第二个0表示的是 key出现的次数 a :0 + 1+ 1  b: 0 +1

        由于我需要得到 value总和 和 key出现的次数。所有需要定义一个元组，初始为0 (0,0)
     */
    //算出value的总和  和 key的次数
    /*
      1：  ("a", 1), ("a", 2), ("b", 3),  (a,(0,0)), (b,(0,0))
      2：  ("b", 4), ("b", 5), ("a", 6) , (a,(0,0)), (b,(0,0))
    分区内：
      1: a : (0,0) + (1,1) + (2,1)  =  (3,2)
        b : (0,0) + (3,1)  = (3,1)
      2: a : (0,0) + (6,1)  = (6,1)
         b:  (0,0) + (4,1) + (5,1) = (9,2)

    分区间： 第一个分区的a  + 第二个分区的a  a (3,2) + (6,1) = (9,3)
            第一个分区的b  + 第二个分区的b  b (3,1) + (9,2) = (12,3)
        */
      val resRdd26: RDD[(String, (Int, Int))] = rdd17.aggregateByKey((0, 0))( //计算的规则  xxxxByKey 是在程序内部进行计算Key，不需要我们去关心。相同Key里面的值进行计算
                                    (t, v) => { //分区内的规则  ： 已经相同key的数据已经进行分组  ("a", 1), ("a", 2),
                                      /* 在当前分区计算 value的总和 和 key 出现的次数
                                        第一分区：   ("a", 1), ("a", 2), ("b", 3)
                                            a (v的总和：3,key次数：2) b(v的总和：3,key次数：1)
                                        第二分区：  ("b", 4), ("b", 5), ("a", 6)
                                            a (v的总和：6,key次数：1)  b(的总和：9,key次数：2)
                                            */
                                      //(0 + 1   , 0 +1  ) ==> a: (0,0) + (1,1)==>a:(1,1)   a:(1,1) + (2,1) ==> a: (3,2)
                                      //(0 +3   , 0 + 1) == > b: (0,0) + (3,1)  ==> b:(3,1)   完成第一分区的计算
                                      //(0 +4  ,  0+1)  ==>  b: (0,0) + (4,1)==> b: (4,1)   b:(4,1) + (5,1) ==> b:(9,2)
                                      //(0 +6   , 0+1  )   ==> a :(0,0)  + (6,1)==> a :(6,1) 完成第二分区的计算
                                      (t._1 + v, t._2 + 1)
                                    },
                                    (t1, t2) => { //分区间的规则  ：  a (v的总和：3,key次数：2)   +    a (v的总和：6,key次数：1)    b(的总和：3,key次数：1) +   b(的总和：9,key次数：2)
                                      //t1:第一分区  t2：第二分区
                                      // a: (3,2) +  a :(6,1)   = a: (9,3)     b:(3,1) + b:(9,2) = b:(12,3)
                                      (t1._1 + t2._1, (t1._2 + t2._2)) // Key：a  value:(9,3)   key：b value:(12,3)

                                    }
                                  )
    //计算平均值 (a,(9,3) )  (b,(12,3) )
    var resRdd27: RDD[(String, Int)] = resRdd26.mapValues {
          // 总和  次数
      case (total, count) => {
        total / count  //  总和 / 次数
      }
    }
    resRdd27.collect().foreach( println )

    resRdd27 = resRdd26.mapValues(item => { //item表示是一个元组，(9,3)   (12,3)
      item._1 / item._2
    })

    resRdd27.collect().foreach( println )


    //15.sortByKey:根据Key来进行排序
    val rdd18: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 1), ("c", 3), ("b", 2)))
    val resRdd28: RDD[(String, Int)] = rdd18.sortByKey() //默认是true 升序 false降序
    resRdd28.collect().foreach( println )
    /*
      join leftOuterJoin RightOuterJoin
     */
    val rdd19: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 1), ("a", 2), ("c", 3), ("b", 8)))
    val rdd20: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 5), ("c", 6), ("a", 3), ("g", 7)))

    val resRdd29: RDD[(String, (Int, Int))] = rdd19.join(rdd20)
    resRdd29.collect().foreach( println )
    /*
    (a,(1,5))
    (a,(1,3))
    (a,(2,5))
    (a,(2,3))
    (c,(3,6))
     */
    val resRdd30: RDD[(String, (Int, Option[Int]))] = rdd19.leftOuterJoin(rdd20)
    resRdd30.collect().foreach( println )


    val resRdd31: RDD[(String, (Option[Int], Int))] = rdd19.rightOuterJoin(rdd20)
    resRdd31.collect().foreach( println )

    //16.cogroup：分组后拼接 （connect + group）
    val rdd21: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 1), ("b", 2)))
    val rdd22: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 4), ("b", 5), ("c", 6),("a",7)))
    val resRdd32: RDD[(String, (Iterable[Int], Iterable[Int]))] = rdd21.cogroup(rdd22)
    resRdd32.collect().foreach( println )
    /* (a ,(1,seq(4,7) ))
       (b,(2,5))
       (c,(3,6))
      拼接后，同一个rdd相同key的数据会放到同一个序列当中
      如果主rdd不存在某个key，但是副rdd存在该键，拼接后然后显示该Key，但是会有一个空序列
     */
    sparkContext.stop()

  }
    // 遍历打印RDD的方法
  def printRDD(rdd:RDD[Int]): Unit ={
    rdd.collect().foreach( println)
  }

}
